package cep;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

public class TestCep04 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> ds = env.readTextFile("data/cep_03.txt");
        Pattern<String, String> pattern = Pattern.<String>begin("start").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.startsWith("A");
            }
        }).oneOrMore().until(new IterativeCondition<String>() {
            @Override
            public boolean filter(String s, Context<String> context) throws Exception {
                System.out.println("^^^^^^^^^^^^^^^^^");
                Iterable<String> iterable = context.getEventsForPattern("start");
                iterable.forEach(t-> System.out.println(t));
                System.out.println("vvvvvvvvvvvvvvvvv");
                return false;
            }
        });

        PatternStream<String> ps = CEP.pattern(ds, pattern);

        ps.flatSelect(new PatternFlatSelectFunction<String, String>() {
            @Override
            public void flatSelect(Map<String, List<String>> map, Collector<String> collector) throws Exception {
                List<String> list = map.get("start");
//                System.out.println("^^^^^^^^^^^^^^^^^");
//                list.forEach(t-> System.out.println(t));
//                System.out.println("vvvvvvvvvvvvvvvvv");
                for (String s : list) {
                    collector.collect(s);
                }
            }
        }).print("$$");



        env.execute();
    }
}
